package kafka

import (
	"context"

	"gitee.com/night-tc/gobige/common"
	"gitee.com/night-tc/gobige/logger"
	kafkago "github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/plain"
)

type option func(*KafkaSendMgr)

func SetKafkaUser(user, pwd string) option {
	return func(mgr *KafkaSendMgr) {
		if user == "" || pwd == "" {
			return
		}
		mgr.Writer.Transport = &kafkago.Transport{
			SASL: plain.Mechanism{
				Username: user,
				Password: pwd,
			},
		}
	}
}

/*
根据目标addr,传输数据的SendMgr
*/
type KafkaSendMgr struct {
	Writer kafkago.Writer
	ctx    context.Context
	cel    context.CancelFunc
	topics []string
	addrs  []string
}

func NewKafkaMgr(topics []string, addrs []string, compf func(messages []kafkago.Message, err error), opts ...option) (result *KafkaSendMgr) {
	result = new(KafkaSendMgr)
	result.topics = topics
	result.addrs = addrs
	if compf == nil {
		compf = result.Completion
	}
	result.Writer = kafkago.Writer{
		Addr:        kafkago.TCP(result.addrs...),
		Async:       true,
		Completion:  compf,
		Logger:      kafkago.LoggerFunc(result.DebugLogger),
		ErrorLogger: kafkago.LoggerFunc(result.ErrorLogger),
		Balancer:    &kafkago.LeastBytes{},
	}
	for _, opt := range opts {
		opt(result)
	}
	return
}

func (this *KafkaSendMgr) Send(index int, key string, value []byte) {
	msg := kafkago.Message{
		Topic: this.topics[index],
		Key:   common.StringToBytes(key),
		Value: value,
	}
	err := this.Writer.WriteMessages(this.ctx, msg)
	if err != nil {
		this.Writer.Completion([]kafkago.Message{msg}, err)
	}
}

func (this *KafkaSendMgr) SendWithTopic(topic string, key string, value []byte) {
	msg := kafkago.Message{
		Topic: topic,
		Key:   common.StringToBytes(key),
		Value: value,
	}
	err := this.Writer.WriteMessages(this.ctx, msg)
	if err != nil {
		this.Writer.Completion([]kafkago.Message{msg}, err)
	}
}

func (this *KafkaSendMgr) Start(ctx context.Context) {
	this.ctx, this.cel = context.WithCancel(ctx)
}

func (this *KafkaSendMgr) Stop() {
	this.Writer.Close()
	this.cel()
}

// 失败的回调
func (this *KafkaSendMgr) Completion(messages []kafkago.Message, err error) {
	if err == nil {
		return
	}
}

func (this *KafkaSendMgr) ErrorLogger(format string, param ...interface{}) {
	logger.Errorf(format, param...)
}

func (this *KafkaSendMgr) DebugLogger(format string, param ...interface{}) {
	logger.Debugf(logger.LogKey_LogKafka, format, param...)
}
